package cn.itcast.streaming.sink;

import cn.itcast.streaming.utils.DateUtil;
import cn.itcast.streaming.utils.StringUtil;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.quotas.AverageIntervalRateLimiter;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.slider.providers.agent.AgentProviderService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class TripDriveSampleToHBaseSink extends RichSinkFunction<String[]> {
    private final static Logger logger = LoggerFactory.getLogger(TripDriveSampleToHBaseSink.class);
    private Table table;
    private Connection conn;
    private BufferedMutator bufferedMutator;
    private String tableName;
    private String cf = "cf";
    public TripDriveSampleToHBaseSink(String tableName) {
        this.tableName = tableName;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        org.apache.hadoop.conf.Configuration conf = HBaseConfiguration.create();
        this.conn = ConnectionFactory.createConnection(conf);
        BufferedMutatorParams bufferedMutatorParams = new BufferedMutatorParams(TableName.valueOf(tableName));
        this.bufferedMutator = conn.getBufferedMutator(bufferedMutatorParams);
        System.out.println("连接hbase对象初始化完成");

    }

    @Override
    public void close() throws Exception {
        super.close();
        if(bufferedMutator!=null) bufferedMutator.close();
        if(conn!=null && !conn.isClosed()) conn.close();

    }

    @Override
    public void invoke(String[] value, Context context) throws Exception {
        Put put = getPut(value);
        bufferedMutator.mutate(put);
        bufferedMutator.flush();
    }

    private Put getPut(String[] value) {
        String rowkey = value[0]+ StringUtil.reverse(value[1]);
        Put put = new Put(Bytes.toBytes(rowkey));
        put.addColumn(Bytes.toBytes(cf),Bytes.toBytes("vin"),Bytes.toBytes(value[0]));
        put.addColumn(Bytes.toBytes(cf),Bytes.toBytes("soc"),Bytes.toBytes(value[2]));
        put.addColumn(Bytes.toBytes(cf),Bytes.toBytes("mileage"),Bytes.toBytes(value[3]));
        put.addColumn(Bytes.toBytes(cf),Bytes.toBytes("speed"),Bytes.toBytes(value[4]));
        put.addColumn(Bytes.toBytes(cf),Bytes.toBytes("gps"),Bytes.toBytes(value[5]));
        put.addColumn(Bytes.toBytes(cf),Bytes.toBytes("terminalTime"),Bytes.toBytes(value[6]));
        put.addColumn(Bytes.toBytes(cf),Bytes.toBytes("processTime"),Bytes.toBytes(DateUtil.getCurrentTime()));
        return put;
    }
}
